package org.xx.armory.spring5.mq;

import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;

import static org.xx.armory.commons.Validators.rejectIfBlank;
import static org.xx.armory.commons.Validators.rejectIfNonPositive;
import static org.xx.armory.commons.Validators.rejectIfNull;

public class RedisMessageQueue
        implements MessageQueue {
    private final StringRedisTemplate stringRedisTemplate;
    private final String queueName;

    public RedisMessageQueue(
            StringRedisTemplate stringRedisTemplate,
            String queueName
    ) {
        this.stringRedisTemplate = rejectIfNull(stringRedisTemplate, "stringRedisTemplate");
        this.queueName = rejectIfBlank(queueName, "queueName");
    }

    @Override
    public boolean produce(
            Supplier<Long> producer
    ) {
        rejectIfNull(producer, "producer");
        final var id = producer.get();
        if (id != null) {
            return this.stringRedisTemplate.boundListOps(this.queueName).leftPush(new MessageQueueId(id).toString()) != null;
        } else {
            return false;
        }
    }

    @Override
    public boolean consume(
            Function<MessageQueueId, Boolean> consumer,
            int readTimeout,
            int messageTimeout
    ) {
        rejectIfNull(consumer, "consumer");
        rejectIfNonPositive(readTimeout, "readTimeout");
        rejectIfNonPositive(messageTimeout, "messageTimeout");

        final var s = this.stringRedisTemplate.boundListOps(this.queueName).rightPop(readTimeout, TimeUnit.SECONDS);
        if (s == null || s.isBlank()) {
            return false;
        }

        final var mqId = MessageQueueId.parse(s);
        if (mqId.isEmpty()) {
            return false;
        }

        final var now = new Date();
        if (now.getTime() - mqId.getTimestamp().getTime() > messageTimeout * 1000L) {
            // 抛弃已超时的消息。
            throw new MessageTimeoutException(mqId.getId(), mqId.getTimestamp(), now);
        }

        final var result = consumer.apply(MessageQueueId.parse(s));

        if (!result) {
            // 未消费的消息，需要被重新推回队列。
            this.stringRedisTemplate.boundListOps(this.queueName).leftPush(s);
        }

        return result;
    }

    @Override
    public void reset() {
        this.stringRedisTemplate.delete(this.queueName);
    }
}
